<?php


namespace App\Libraries\Queue;


use Closure;
use Exception;
use Illuminate\Support\Facades\Log;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use Throwable;

class AMQPConsumer
{
    /** @var null */
    private static $instance = null;
    /** @var AMQPStreamConnection */
    private $connection;
    /** @var \PhpAmqpLib\Channel\AMQPChannel */
    private $channel;
    /** @var array */
    private $connectionConfig = [];
    /** @var HandleMessage  */
    private $callback = null;

    /**
     * AMQPConsumer constructor.
     * @param string $host
     * @param int $port
     * @param string $user
     * @param string $password
     * @param string $vHost
     * @author luffyzhao@vip.126.com
     */
    private function __construct(string $host, int $port, string $user, string $password, string $vHost = '/')
    {
        $this->connectionConfig = [
            'host' => $host,
            'port' => $port,
            'user' => $user,
            'password' => $password,
            'vHost' => $vHost,
        ];
        $this->connect();
    }

    /**
     * @author luffyzhao@vip.126.com
     */
    private function connect()
    {
        $this->connection = new AMQPStreamConnection($this->connectionConfig['host'], $this->connectionConfig['port'],
            $this->connectionConfig['user'], $this->connectionConfig['password'], $this->connectionConfig['vHost']);
        $this->channel = $this->connection->channel();
    }

    /**
     * @param string $host
     * @param int $port
     * @param string $user
     * @param string $password
     * @param string $vHost
     * @return AMQPConsumer
     * @author luffyzhao@vip.126.com
     */
    public static function instance(string $host, int $port, string $user, string $password, string $vHost = '/')
    {
        if (self::$instance === null) {
            self::$instance = new static($host, $port, $user, $password, $vHost);
        }
        return self::$instance;
    }

    /**
     * @param $queueName
     * @param $exchangeName
     * @param $exchangeType
     * @param $routeKey
     * @author luffyzhao@vip.126.com
     */
    protected function setConfig($queueName, $exchangeName, $exchangeType, $routeKey)
    {
        $this->channel->exchange_declare($exchangeName, $exchangeType, false, true, false);
        $this->channel->queue_declare($queueName, false, true, false, false);
        $this->channel->queue_bind($queueName, $exchangeName, $routeKey);
    }

    /**
     * @param $queueName
     * @param $exchangeName
     * @param $exchangeType
     * @param $routeKey
     * @throws \ErrorException
     * @author luffyzhao@vip.126.com
     */
    public function doProcess($queueName, $routeKey, $exchangeName = 'amq.direct', $exchangeType = 'direct')
    {
        $this->setConfig($queueName, $exchangeName, $exchangeType, $routeKey);
        $this->channel->basic_consume($queueName, '', false, false, false, false, function (AMQPMessage $msg) {
            try {
                if($this->callback instanceof HandleMessage){
                    $this->callback->handle($msg);
                }
                $msg->getChannel()->basic_ack($msg->getDeliveryTag());
            } catch (Exception | Throwable $exception) {
                Log::error($exception->getMessage());
            }
        });

        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
    }

    /**
     * @param $callback
     * @author luffyzhao@vip.126.com
     */
    public function doCallback(HandleMessage $callback)
    {
        $this->callback = $callback;
    }

    /**
     * @param $body
     * @param $queueName
     * @param $exchangeName
     * @param $exchangeType
     * @param $routeKey
     * @author luffyzhao@vip.126.com
     */
    public function publisher($body, $queueName, $routeKey, $exchangeName = 'amq.direct', $exchangeType = 'direct')
    {
        $this->setConfig($queueName, $exchangeName, $exchangeType, $routeKey);
        $message = new AMQPMessage($body, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
        $this->channel->basic_publish($message, $exchangeName, $routeKey);
    }

    /**
     * @author luffyzhao@vip.126.com
     */
    private function __clone()
    {
    }

    /**
     * @throws Exception
     */
    public function __destruct()
    {
        $this->channel->close();
        $this->connection->close();
    }
}